package com.gitee.transformation

import com.gitee.source.practice
import com.gitee.source.practice.CustomerSourceMySQL
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}

/*
  分组: 批处理中使用groupBy, 流处理中使用keyBy
*/
object KeyByTest {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._
    //获取mysql数据
    val data: DataStream[practice.Product] = CustomerSourceMySQL.mysqlConsumer(senv)
    //处理数据
    val pnameData: DataStream[Char] = data.map(pro => {
      (pro.pid, pro.pname, pro.num)
    }).flatMap(iter => {
      iter._2
    })
    val pnameOne: DataStream[(Char, Int)] = pnameData.map((_,1))
    //分组: 批处理中使用groupBy, 流处理中使用keyBy
    val groupPname: KeyedStream[(Char, Int), Tuple] = pnameOne.keyBy(0)
    val result: DataStream[(Char, Int)] = groupPname.sum(1)
    result.print()
    senv.execute()
  }

}
